package my.thesis.bolts.fieldExtractorsReliable;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import my.thesis.analytics.metrics.ThroughputMetric;

import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class TweetFieldExtractorTopo2Reliable extends BaseRichBolt  
{
			private static final long serialVersionUID = 1L;	
			private OutputCollector collector;
				
			transient ThroughputMetric _throughputMetric;
			private static Boolean metricsOn=false;
			
			
			public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
			{
				this.collector=collector;
				
				if(metricsOn)
				{
					_throughputMetric=new ThroughputMetric();
					context.registerMetric("tupleCount", _throughputMetric, 30);
					_throughputMetric.setStartTime();
				}
			}
				
			public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
			{		
					outputFieldsDeclarer.declareStream("textStream", new Fields("id","millis","country","lang","text","lat","lon"));   
					outputFieldsDeclarer.declareStream("influenceStream", new Fields("id","username","mention","followers","text")); 
			}
			
			public void execute(Tuple tuple) 
			{
				if(metricsOn)
				{
					_throughputMetric.incrTuples();
				}
				
				Set<String> mentions = new HashSet<String>();

				String jsonString = tuple.getStringByField("jsonString");
				
				JSONObject obj = null;
				try
				{
					obj = new JSONObject(jsonString);
				}
				catch(Exception ex)
				{
					System.out.println("Something went wrong...");
				}
				
				if(obj != null)
				{
					Long id = obj.getLong("id");
					String username=obj.getString("username");
					int followers = obj.getInt("followers");
					
					String lang = obj.getString("lang");
					
					String text = obj.getString("text");
					String country = obj.getString("country");
					
					DateTime timestamp=new DateTime(obj.get("createdat"));
					Long millis = timestamp.getMillis();
					
					//TEST
					//DateTime timestamp = DateTime.now();
					//Long millis = timestamp.getMillis();
					//TEST
					
					Double  lat = obj.getDouble("lat");
					Double lon = obj.getDouble("lon");
				
					if(obj.has("mentions"))
					{
						JSONArray mentionsArray = obj.getJSONArray("mentions");
						for (int i=0; i < mentionsArray.length(); i++)
						{
							mentions.add(mentionsArray.getJSONObject(i).getString("name"));
						}
					}
					
					collector.emit("textStream", tuple, new Values(id,millis,country,lang,text,lat,lon));
					
					for (String mention : mentions)
					{
						collector.emit("influenceStream",tuple,new Values(id,username,mention,followers,text));
					}	
				
					collector.ack(tuple);
				}
				else
				{
					collector.fail(tuple);
				}		
			}		
}
